<?php

include __DIR__ .'/mqtt/examples/bootstrap.php';

use Simps\MQTT\Tools\Common;
use Simps\MQTT\Protocol\Types;
use Simps\MQTT\Protocol\V3;

use Swoole\Coroutine;
use Swoole\Database\PDOConfig;
use Swoole\Database\PDOPool;
use Swoole\Database\RedisConfig;
use Swoole\Database\RedisPool;
use Swoole\Process;

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

use lib\core\error;
use lib\amqp\amqpPool;
use lib\redis\redisLib;

Coroutine::set(['hook_flags'=> SWOOLE_HOOK_ALL]);

class MqttServer
{
    public $server;

    private $topicPrefix = "simps-mqtt/chargeChamber/";

    //Redis连接池
    private $redisPool = null;

    //Mysql连接池
    private $mysqlPool = null;

    //配置文件
    private $config = null;

    //mq连接池
    private $mqPool = null;

    //消费者
    private $subrible;

    //获取本机ip信息
    private $ip;

    public function __construct()
    {
        //psr4加载
        require_once "lib/core/load.php";
        lib\core\load::register();

        //自定义异常
        error::register();

        //载入配置文件
        $this->autoload();

        $this->ip = $ip = swoole_get_local_ip();

        $this->server = new Swoole\Server(
            '0.0.0.0',
            9616,
            SWOOLE_PROCESS
        );

        /**
         * 配置
         */
        $this->server->set([
            'open_mqtt_protocol' => $this->config['OPEN_MQTT_PROTOCOL'],
            'worker_num' => $this->config['WORKER_NUM'],
            'package_max_length' => $this->config['PACKAGE_MAX_LENGTH'],
            'daemonize' => $this->config['DAEMONIZE'],
            'log_file' => __DIR__ .'/logs/iot_'.date('Y-m-d').'.log',
            /**
             * SWOOLE_LOG_DEBUG 调试日志，仅作为内核开发调试使用
             * SWOOLE_LOG_TRACE 跟踪日志，可用于跟踪系统问题，调试日志是经过精心设置的，会携带关键性信息
             * SWOOLE_LOG_INFO  普通信息，仅作为信息展示
             * SWOOLE_LOG_NOTICE        提示信息，系统可能存在某些行为，如重启、关闭
             * SWOOLE_LOG_WARNING       警告信息，系统可能存在某些问题
             * SWOOLE_LOG_ERROR 错误信息，系统发生了某些关键性的错误，需要即时解决
             * SWOOLE_LOG_NONE  相当于关闭日志信息，日志信息不会抛出
             */
            'LOG_LEVEL' => $this->config['LOG_LEVEL'],
            'heartbeat_idle_time' => $this->config['HEARTBEAT_CHECK_INTERVAL'],
            'heartbeat_check_interval' => $this->config['HEARTBEAT_IDLE_TIME'],
            'max_coroutine' => $this->config['MAX_COROUTINE'],
            'max_wait_time' => $this->config['MAX_WAIT_TIME'],
            'reload_async' => $this->config['RELOAD_ASYNC'],
        ]);

        //设置监听
        $this->server->on('start', [$this, 'onStart']);
        $this->server->on('workerstart', [$this, 'onWorkerStart']);
        $this->server->on('connect', [$this, 'onConnect']);
        $this->server->on('receive', [$this, 'onReceive']);
        $this->server->on('close', [$this, 'onClose']);

        if($this->config["CLUSTER_MODE"]) { 

            //订阅
            $this->subrible = new Swoole\Process(function ($process) {

                $socket = $process->exportSocket();
                $connection = new AMQPStreamConnection($this->config["rabbitmq"]["RABBITMQ_SERVER_IP"], $this->config["rabbitmq"]["RABBITMQ_SERVER_PORT"], $this->config["rabbitmq"]["RABBITMQ_SERVER_USER"], $this->config["rabbitmq"]["RABBITMQ_SERVER_PASS"], $this->config["rabbitmq"]["RABBITMQ_SERVER_VHOST"]);
                $channel = $connection->channel();
                $channel->exchange_declare($this->config["rabbitmq"]["RABBITMQ_SERVER_FANOUT_EXCHANGE_NAME"], 'fanout', false, true, false);
                //防止系统崩溃, 所有消息丢失, 第三个参数设置为true持久化
                // $channel->queue_declare("test-queue-1", false, false, false, false);
                // 声明队列，如果队列名为空，则自动生成一个唯一ID，且返回队列名
                list($queue_name, ,) = $channel->queue_declare(
                    "", // 队列名，不允许重复，如果为空则自动生成一个唯一ID，这个时候就是一个匿名队列
                    false,
                    false, // 是否持久化
                    false,
                    false
                );
                $channel->queue_bind($queue_name, $this->config["rabbitmq"]["RABBITMQ_SERVER_FANOUT_EXCHANGE_NAME"]);
                $callback = function ($msg) {
                    echo ' [x] Received ', $msg->body, "\n";
                    $data = json_decode($msg->body, true);
                    if(!empty($data)) {

                        if($this->ip["eth0"] == $data["ip"]) {

                            echo 'has been filter';
                        }else {

                            $redis = redisLib::getInstance($this->config["redis"]);
                            $connections = $redis->hget("iot", $this->topicPrefix.$data["device"]);
                            if(!empty($connections)) {
                                foreach ($connections as $conn) {
                                    $this->server->send($conn, V3::pack[$data]);
                                }
                            }

                            //消息确认
                            $msg->ack();
                            //判断获取到quit后
                            if (trim($msg->body) == 'quit') {
                                var_dump('quit: '.$msg->getConsumerTag());
                                $msg->getChannel()->basic_cancel($msg->getConsumerTag());
                            }
                        }
                    }
                };

                /**
                 * 公平调度
                 * 等待消费者处理完成后再接收新消息，不堆积
                 */
                //$channel->basic_qos(null, 1, null);

                $channel->basic_consume($queue_name, '', false, false, false, false, $callback);

                while ($channel->is_consuming()) {
                    $channel->wait();
                }
            }, false, 2, 1);
            $this->server->addProcess($this->subrible);
        }

        //启动服务
        $this->server->start();
    }

    /**
     * 启动后在主进程（master）的主线程回调此函数
     */
    public function onStart(Swoole\Server $server){

        $this->debug("onStart...\n");

        /**
         * 记录master和manager的pid
         * 重启所有worker进程 kill -USR1 主进程PID
         * 仅重启task进程 kill -USR2 主进程PID
         */
        file_put_contents(__DIR__ ."/pid/master.pid", $server->master_pid);
        file_put_contents(__DIR__ ."/pid/manager.pid", $server->manager_pid);
    }

    /*
     * WorkerStart 启动
     */
    function onWorkerStart(Swoole\Server $server, $worker_id){

        $this->debug("已开启worker".$worker_id.", pid: ".$server->worker_pid."\n");

        $this->mysqlPool = new PDOPool((new PDOConfig)
            ->withHost($this->config['MYSQL_SERVER_HOST'])
            ->withPort($this->config['MYSQL_SERVER_PORT'])
            ->withDbName($this->config['MYSQL_DB_NAME'])
            ->withCharset($this->config['MYSQL_SERVER_CHARSET'])
            ->withUsername($this->config['MYSQL_SERVER_USER'])
            ->withPassword($this->config['MYSQL_SERVER_PASS'])
        ,$this->config['MYSQL_CONNECT_SIZE']);

        $this->redisPool = new RedisPool((new RedisConfig)
            ->withHost($this->config['redis']['REDIS_SERVER_IP'])
            ->withPort($this->config['redis']['REDIS_SERVER_PORT'])
            ->withAuth($this->config['redis']['REDIS_SERVER_AUTH'])
            ->withDbIndex($this->config['redis']['REDIS_SERVER_DB'])
            ->withTimeout($this->config['redis']['REDIS_SERVER_TIMEOUT'])
        ,$this->config['redis']['REDIS_CONNECT_SIZE']);

        $this->mqPool = amqpPool::getInstance();
        $this->mqPool->initConfig($this->config["rabbitmq"]);

        //var_dump(get_included_files()); //此数组中的文件表示进程启动前就加载了，所以无法reload
    }

    /**
     * 当 Mqtt 客户端与服务器建立连接并完成握手后会回调此函数。
     */
    public function onConnect(Swoole\Server $server, $fd){

        $this->debug("socket连接".$fd.", Client #{$fd}: Connect.\n");
    }

    /**
     * 当服务器收到来自客户端的数据帧时会回调此函数。
     */
    public function onReceive(Swoole\Server $server, $fd, $from_id, $data){

        try {

            $data = V3::unpack($data);
            if (is_array($data) && isset($data['type'])) {

                if(isset($data["user_name"]) && isset($data["password"])){

                    if($data["user_name"] != "chargechamber-iot" or $data["password"] != "BcNTY0ZjwQjqtTaU"){

                        $server->close($fd);
                    }
                }

                $this->debug(date("Y-m-d H:i:s")."socket连接: ".$fd.", 已接收数据: \n");
		        $this->debug(json_encode($data));

                switch ($data['type']) {
                    case Types::CONNECT:
                        // Check protocol_name
                        if ($data['protocol_name'] != 'MQTT') {

                            $server->close($fd);
                        }else{

                            $server->send(
                                $fd,
                                V3::pack(
                                    [
                                        'type' => Types::CONNACK,
                                        'code' => 0,
                                        'session_present' => 0,
                                    ]
                                )
                            );
                        }

                        break;
                    case Types::PINGREQ:
                        $server->send($fd, V3::pack(['type' => Types::PINGRESP]));
                        break;
                    case Types::DISCONNECT:
                        if ($server->exist($fd)) {
                            $server->close($fd);
                        }
                    break;
                    case Types::PUBLISH:

                        $message = json_decode($data["message"], true);

                        /**
                         * 幂等解决方案
                         * 1.唯一索引
                         * 2.token机制 建议使用
                         * 3.悲观锁
                         * 4.乐观锁
                         * 5.分布式锁
                         */
                        //code...

                        if ($data['qos'] == 1) {
                            $server->send(
                                $fd,
                                V3::pack(
                                    [
                                        'type' => Types::PUBACK,
                                        'message_id' => $data['message_id'] ?? '',
                                    ]
                                )
                            );
			            }

                        //发送订阅的用户
                        $redis = $this->redisPool->get();
                        $connections = $redis->hget("iot", $this->topicPrefix.$message["device"]);
                        if(!empty($connections)) {
                            foreach($connection as $conn) {
                        
                                $this->server->send(
                                    $conn, 
                                    V3::pack(
                                        [
                                            'type' => $data['type'],
                                            'topic' => $data['topic'],
                                            'message' => [],
                                            'dup' => $data['dup'],
                                            'qos' => $data['qos'],
                                            'retain' => $data['retain'],
                                            'message_id' => $data['message_id'] ?? ''
                                        ]
                                ));
                            }
                        }
                        $this->redisPool->put($redis);

                        //发送mq
                        $this->sendMq(array(
                            'ip' => $this->ip["eth0"],
                            'data' => $data,
                            'device' => $message["device"]
                        ));
                    break;
                    case Types::SUBSCRIBE:

                        //订阅
                        $redis = $this->redisPool->get();

                        $payload = [];
                        foreach ($data['topics'] as $k => $qos) {

                            $this->debug(date("Y-m-d H:i:s").", 订阅: socket连接：".$fd."已订阅, type: ".Types::SUBSCRIBE."\n");

                            $redis->hset("iot", $k, $fd);

                            if (is_numeric($qos) && $qos < 3) {
                                $payload[] = $qos;
                            } else {
                                $payload[] = 0x80;
                            }
                            $payload[] = 1;
                        }

                        $server->send(
                            $fd,
                            V3::pack(
                                [
                                    'type' => Types::SUBACK,
                                    'message_id' => $data['message_id'] ?? '',
                                    'payload' => $payload,
                                ]
                            )
                        );
                        $this->redisPool->put($redis);
                        break;
                    case Types::UNSUBSCRIBE:
                        $server->send(
                            $fd,
                            V3::pack(
                                [
                                    'type' => Types::UNSUBACK,
                                    'message_id' => $data['message_id'] ?? '',
                                ]
                            )
                        );
                        break;
                }
            } else {
                $server->close($fd);
            }
        } catch (\Throwable $e) {
            $this->debug(date("Y-m-d H:i:s")." 错误errorThrowable \033[0;31mError: {$e->getMessage()}\033[0m\r\n");
            $server->close($fd);
        }
    }

    public function sendMq($msgArr = array()) {

	    if($this->config["CLUSTER_MODE"]) {

            $conn = $this->mqPool->getConnection();
            $c = $conn["db"];

            //推送成功
            $c->set_ack_handler(
                function (AMQPMessage $message) {
                    echo "成功: " . $message->body . PHP_EOL;
                }
            );

            //推送失败
            $c->set_nack_handler(
                function (AMQPMessage $message) {
                    echo "失败: " . $message->body . PHP_EOL;
                }
            );
            $c->confirm_select();
            $c->exchange_declare($this->config["rabbitmq"]["RABBITMQ_SERVER_FANOUT_EXCHANGE_NAME"], 'fanout', false, true, false);
            $msg = new AMQPMessage(json_encode($msgArr));
            $c->basic_publish($msg, $this->config["rabbitmq"]["RABBITMQ_SERVER_FANOUT_EXCHANGE_NAME"]);
            $c->wait_for_pending_acks();
            $this->mqPool->free($conn);
	    }
    }

    //进程退出清理某些长连接 Socket
    public function onWorkerExit(Swoole\Server $server, $worker_id){
        
    }

    public function onWorkerStop(Swoole\Server $server, $worker_id){

        $this->debug(date("Y-m-d H:i:s")." worker已关闭, worker_id: ".$worker_id."\n");
    }

    /**
     * 关闭fd
     */
    public function onClose($ser, $fd){

        $this->debug(date("Y-m-d H:i:s")." 主动断开, socket连接: ".$fd.", client {$fd} closed\n");
        $this->cleanFdInfo($fd);
    }

    private function cleanFdInfo($fd = 0) {
    
        $this->debug("下线fd: ".$fd."\n");
    }

    private function autoload(){

        if (is_null($this->config)) {

            //加载配置文件
            $this->config = require_once "config.php";
        }
    }

    //debug
    private function debug($str){

        if ($this->config['DEBUG']) {
            echo "-------------------------------\n";
            echo '[' . date("Y-m-d H:i:s") . '] Debug:['. $str . "]\n";
            echo "-------------------------------\n";
        }
    }
}

new MqttServer();
